package cn.doitedu.demo4.functions;

import cn.doitedu.demo4.calculators.RuleCalculator;
import cn.doitedu.demo4.beans.EventBean;
import cn.doitedu.demo4.beans.RuleMetaBean;
import cn.doitedu.demo4.utils.RuleCalculatorBuilder;
import cn.doitedu.demo4.utils.StateDescUtil;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class RuleProcessCoreFunction extends KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String> {

    ConcurrentHashMap<String, RuleCalculator> calculatorPool = new ConcurrentHashMap<>();

    GroovyClassLoader groovyClassLoader;
    MapState<String, String> oldCalculatorIdsMapState;
    ListState<EventBean> recentEventState;

    @Override
    public void open(Configuration parameters) throws Exception {

        groovyClassLoader = new GroovyClassLoader();

        // 用于记录哪些运算机是已经处理过数据的老运算机
        oldCalculatorIdsMapState = getRuntimeContext().getMapState(StateDescUtil.getOldCalculatorIdsDesc());

        // 用于缓存最近 10秒钟event数据的list状态
        recentEventState = getRuntimeContext().getListState(StateDescUtil.getRecentEventsListStateDesc());


    }

    @Override
    public void processElement(EventBean eventBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {


        ReadOnlyBroadcastState<String, RuleMetaBean> broadcastState = ctx.getBroadcastState(StateDescUtil.getMetaBcStateDesc());

        /**
         * 1.故障恢复逻辑
         */
        if(calculatorPool.isEmpty()){
            // 如果运算机池为空，则说明系统要么是第一次刚启动，要么是故障后刚重启
            // 要重建故障重启前所有的运算机对象
            RuleCalculatorBuilder.restoreAllCalculator(broadcastState,groovyClassLoader,getRuntimeContext(),calculatorPool);
        }


        /**
         * 2.正常逻辑
         */
        // 遍历运算机池的每一个运算机
        for (Map.Entry<String, RuleCalculator> entry : calculatorPool.entrySet()) {

            String ruleId = entry.getKey();
            RuleCalculator ruleCalculator = entry.getValue();

            // 判断当前遍历到的运算机，是否是一个新上线的运算机
            if(! oldCalculatorIdsMapState.contains(ruleId)){
                // 老运算机id集合中，不包含该运算机的规则id，说明它是一个新运算机
                // 要先将缓存中的历史 events 对它重放
                for (EventBean bean : recentEventState.get()) {
                    ruleCalculator.calculate(bean,out);
                }

                // 将它的id添加到老运算机id集合
                oldCalculatorIdsMapState.put(ruleId,"");
            }

            // 将当前收到的event,交给运算机处理
            ruleCalculator.calculate(eventBean,out);

            // 将本次收到的event，放入list状态缓存起来
            recentEventState.add(eventBean);

        }
    }

    @Override
    public void processBroadcastElement(RuleMetaBean metaBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.Context ctx, Collector<String> out) throws Exception {

        BroadcastState<String, RuleMetaBean> broadcastState = ctx.getBroadcastState(StateDescUtil.getMetaBcStateDesc());


        /**
         * 1.故障恢复逻辑
         */
        // 判断运算机池是否为空
        if(calculatorPool.isEmpty()){
            // 要重建故障重启前所有的运算机对象
            RuleCalculatorBuilder.restoreAllCalculator(broadcastState,groovyClassLoader,getRuntimeContext(),calculatorPool);
        }


        /**
         * 2. 正常情况下的逻辑
         */
        if(metaBean.getOp().equals("d") || metaBean.getRule_status() == 0 ){
            // 下线操作,从运算机池移除运算机对象
            calculatorPool.remove(metaBean.getRule_id());

            // 从广播状态移除，规则元数据Bean
            broadcastState.remove(metaBean.getRule_id());


            log.warn("下线了一个规则，规则id：{}",metaBean.getRule_id());
        }else{
            // 更新覆盖、新增规则上线操作
            // 构建运算机
            RuleCalculator ruleCalculator = RuleCalculatorBuilder.buildCalculator(metaBean, groovyClassLoader);
            // 初始化
            ruleCalculator.init(getRuntimeContext(), metaBean.getRule_param(),metaBean.getPreSelectUserBitmap(), metaBean.getHistory_end_time());
            // 放入运算机池
            calculatorPool.put(metaBean.getRule_id(),ruleCalculator);

            // 将本次收到的新规则元数据Bean，备份到广播状态
            broadcastState.put(metaBean.getRule_id(),metaBean);

            log.warn("上线了一个规则，规则id：{}",metaBean.getRule_id());

        }
    }
}
